%%% File    : queue_sup.erl
%%% Author  :  <>
%%% Description : 
%%% Created : 29 Mar 2011 by  <>

-module(queue_sup).

-behaviour(gen_server).

-export([start_link/0]).

-export([start_queue_server/1]).
-export([add_extra/1, del_extra/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	terminate/2, code_change/3]).

-record(state, {type,
	       queue_server,   %%[{node, pid, idlist}]
	       extra_node}).

-define(Sup, ?MODULE).

%%-----
start_queue_server(Node) when is_atom(Node) ->
    gen_server:call({global, ?Sup}, {start_queue_server, Node});
start_queue_server(_) ->
    {error, badarg}.

add_extra(Node) when is_atom(Node) ->
    gen_server:call({global, ?Sup}, {add_extra, Node});
add_extra(_) ->
    {error, badarg}.

del_extra(Node) when is_atom(Node) ->
    gen_server:call({global, ?Sup}, {del_extra, Node});
del_extra(_) ->
    {error, badarg}.

start_link() ->
    gen_server:start_link({global, ?Sup}, ?MODULE, [], []).

%%-------
init([]) ->
    process_flag(trap_exit, true),
    pg2:create(queue_server),
    S =#state{type=queue,
	     queue_server=[],
	     extra_node=[]},
    {ok, S}.

handle_call({add_extra, Node}, _From, S=#state{extra_node=Extra}) ->
    case lists:member(Node, Extra) of
	true ->
	    {reply, {error, exist}, S};
	false ->
	    {reply, ok, S#state{extra_node=[Node|Extra]}}
    end;

handle_call({del_extra, Node}, _From, S=#state{extra_node=Extra}) ->
    case list:member(Node, Extra) of
	true ->
	    {reply, ok, S#state{extra_node=lists:delete(Node, Extra)}};
	false ->
	    {reply, {error, noexist}, S}
    end;

handle_call({start_queue_server, Node}, _From, S) ->
    {Res, Ns} = do_start(Node, S),
    {reply, Res, Ns};
    
handle_call(_Req, _From, S) ->
    {reply, ok, S}.

handle_cast({redirect, Node, Pid, Idlist}, S=#state{queue_server=Qserver}) ->
    link(Pid),
    {noreply, S#state{queue_server=[{Node, Pid, Idlist}|Qserver]}};

handle_cast(_Msg, S) ->
    {noreply, S}.

handle_info({'EXIT', _, normal}, S) ->
    {noreply, S};

handle_info({'EXIT', Pid, _}, S=#state{queue_server=Qserver}) ->
    case lists:keysearch(Pid, 2, Qserver) of
	false ->
	    {noreply, S};
	{value, {Node, Pid, Idlist}} ->
	    restart(Node, Idlist, S=#state{queue_server=lists:keydelete(Node, 1, Qserver)})
    end;
    
handle_info(_Info, S) ->
    {noreply, S}.

terminate(_Reason, _S) ->
    ok.
code_change(_Oldvsn, S, _Extra) ->
    {ok, S}.

%%----
do_start(Node, S=#state{queue_server=Qserver}) ->
    case lists:keysearch(Node, 1, Qserver) of
	{value, _} ->
	    {{error, exist}, S};
	false ->
	    Res = rpc:call(Node, queue_server, start, [self(), all_nodes(Qserver), []]),
	    case Res of
		{ok, Pid} ->
		    link(Pid),
		    pg2:join(queue_server, Pid),
		    {ok, S#state{queue_server=[{Node, Pid, []}|Qserver]}};
		Err ->
		    {{error, Err}, S}
            end
    end.
		    
restart(Node, Idlist, S) ->
    restart(Node, Idlist, S, 2).

restart(Node, Idlist, S=#state{extra_node=Extra}, []) ->
    io:format("restart queue_server on node:~p failed~n", [Node]),
    start_on_extra(Idlist, Extra, S);
restart(Node, Idlist, S=#state{queue_server=Qserver}, Try) ->
    Res = rpc:call(Node, queue_server, start, [self(), all_nodes(Qserver), Idlist]),
    case Res of
	{ok, Pid} ->
	    link(Pid),
	    pg2:join(queue_server, Pid),
	    {noreply, S#state{queue_server=[{Node, Pid, Idlist}|Qserver]}};
	_Err ->
	    restart(Node, Idlist, S, Try-1)
    end.

start_on_extra(_Idlist, [], S) ->
    io:format("restart queueserver on extra failed~n", []),
    {noreply, S};
start_on_extra(Idlist, [Ex|Extra], S=#state{queue_server=Qserver}) ->
    Res = rpc:call(Ex, queue_server, start, [self(), all_nodes(Qserver), Idlist]),
    case Res of
	{ok, Pid} ->
	    link(Pid),
	    pg2:join(queue_server, Pid),
	    {noreply, S#state{queue_server=[{Ex, Pid, Idlist}|Qserver],
			     extra_node=Extra}};
	_Err ->
	    start_on_extra(Idlist, Extra, S=#state{extra_node=Extra})
    end.

all_nodes(Qserver) ->
    all_nodes(Qserver, []).

all_nodes([], Nodes) ->
    Nodes;
all_nodes([{Node, _, _}|T], Nodes) ->
    all_nodes(T, [Node|Nodes]).
